package com.atguigu.gmall.realtime.common.util;

import com.atguigu.gmall.realtime.common.constant.Constant;

/**
 * ClassName: SQLUtil
 * Description:
 * FlinkSQL工具类
 *
 * @Create 2024/4/20 14:59
 */
public class SQLUtil {
    //获取kafka连接器的连接属性
    public static String getKafkaDDL(String topic, String groupId) {
        return " WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = '" + topic + "',\n" +
                "  'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "',\n" +
                "  'properties.group.id' = '" + groupId + "',\n" +
                "  'scan.startup.mode' = 'latest-offset',\n" +
                "  'format' = 'json'\n" +
                ")";
    }

    //获取HBase连接器的连接属性
    public static String getHBaseDDL(String tableName) {
        return " WITH ( 'connector' = 'hbase-2.2',\n" +
                "    'table-name' = '" + tableName + "',\n" +
                "    'zookeeper.quorum' = 'hadoop102:2181',\n" +
                "    'lookup.async' = 'true',\n" +
                "    'lookup.cache' = 'PARTIAL',\n" +
                "    'lookup.partial-cache.max-rows' = '500',\n" +
                "    'lookup.partial-cache.expire-after-write' = '1 hour',\n" +
                "    'lookup.partial-cache.expire-after-access' = '1 hour');";
    }

    //获取upsert-kafka连接器的连接属性
    public static String getUpsertKafkaDDL(String topic) {
        return " WITH ( 'connector' = 'upsert-kafka',\n" +
                "    'topic' = '" + topic + "',\n" +
                "    'properties.bootstrap.servers' = '" + Constant.KAFKA_BROKERS + "',\n" +
                "    'key.format' = 'json',\n" +
                "    'value.format' = 'json');";
    }

    //获取doris连接器连接属性
    public static String getDorisSinkDDL(String tableName) {
        return "WITH (" +
                "  'connector' = 'doris', " +
                "  'fenodes' = '" + Constant.DORIS_FE_NODES + "', " +
                "  'table.identifier' = '" + Constant.DORIS_DATABASE + "." + tableName + "', " +
                "  'username' = 'root', " +
                "  'password' = 'aaaaaa', " +
                //要写入的格式
                "  'sink.properties.format' = 'json', " +
                "  'sink.buffer-count' = '4', " +
                "  'sink.buffer-size' = '4086'," +
                "  'sink.enable-2pc' = 'false', " + // 测试阶段可以关闭两阶段提交,方便测试
                "  'sink.properties.read_json_by_line' = 'true' " +
                ") ";
    }
}
